package com.xsomnus.rpc.consumer.netty;

import com.alibaba.fastjson.JSONArray;
import com.xsomnus.rpc.common.codec.FastJsonDecoder;
import com.xsomnus.rpc.common.codec.FastJsonEncoder;
import com.xsomnus.rpc.common.entity.Request;
import com.xsomnus.rpc.common.entity.Response;
import com.xsomnus.rpc.consumer.net.ConnectionManager;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.SocketAddress;
import java.util.concurrent.SynchronousQueue;

/**
 * @author @xsomnus666_xiawenye★
 * @since 2019/4/22 0022 16:42
 * 当你发现自己的才华撑不起野心时，就请安静下来学习吧
 */
@Slf4j
@Component
public class NettyClient {

    private EventLoopGroup group = new NioEventLoopGroup(1);
    private Bootstrap bootstrap = new Bootstrap();

    @Autowired
    NettyClientHandler clientHandler;

    @Autowired
    ConnectionManager connectionManager;

    public NettyClient() {
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new IdleStateHandler(0,0, 30));
                        pipeline.addLast(new FastJsonEncoder());
                        pipeline.addLast(new FastJsonDecoder());
                        pipeline.addLast("handler", clientHandler);
                    }
                });
    }

    /**
     * 发送请求(RPC调用)
     */
    public Object send(Request request) throws InterruptedException {

        Channel channel = connectionManager.chooseChannel();

        if (channel != null && channel.isActive()) {
            SynchronousQueue<Object> queue = clientHandler.sendRequest(request, channel);
            Object ret = queue.take();
            return JSONArray.toJSONString(ret);
        } else {
            Response response = new Response();
            response.setCode(-2);
            response.setErrMsg("未能正确连接到服务器, 请检查相关的配置");
            return JSONArray.toJSONString(response);
        }


    }

    /**
     * 连接远程服务
     */
    public Channel doConnect(SocketAddress address) throws InterruptedException {
        return bootstrap.connect(address).sync().channel();
    }
}
